//
// Created by benxb on 2021/11/29.
//

#include <algorithm>
#include <thread>
#include "storage/data_chunk_io.h"
#include "mem/data_table_manager.h"
#include <common/logger.h>
#include <chrono>
#include <iostream>

namespace benxdb {

void FlushTable(DataTableManager *me) {
    Timer wait_timer, flush_timer;
    while (true) {
        wait_timer.Start();
        std::unique_lock <std::mutex> lck(me->frozen_queue_mtx_);
        while (me->frozen_tables_.empty()) {
            me->flush_cv_.wait(lck);
        }
        MemDataTable* data_table = me->frozen_tables_.front();
        me->frozen_tables_.pop_front();
        lck.unlock();

        if (data_table == nullptr || data_table->IsEmpty()) {
            break;
        }
        wait_timer.PauseAndPrint("Wait table.");

        flush_timer.Start();
        LOG_DEBUG("Flush Table: %lu", data_table->GetLSN());
        for (int i = 0; i < data_table->ColumnSize(); i++) {
            std::shared_ptr<DataChunk> chunk_ptr = data_table->GetChunkByIndex(i);
            me->chunk_io_->AppendDataChunk(chunk_ptr.get(), false);
            auto col = me->table_->GetColumn(i);
            if (col.GetIndexType() == LINEAR_INDEX) {
                uint64_t first_id = data_table->GetLSN() * RECORD_COUNT_PER_CHUNK;
                me->linear_indices_[col.GetName()]->GenerateIndex(chunk_ptr->GetData(), chunk_ptr->GetChunkSize(), first_id);
            }
        }
        me->table_->SetFlushLSN(data_table->GetLSN());
        data_table->Clear();
        me->InsertMemTable(data_table);
        flush_timer.PauseAndPrint("Flush Table");
    }
    wait_timer.StopAndPrint("Wait table total.");
    flush_timer.StopAndPrint("Flush Table total.");
}

DataTableManager::DataTableManager(const std::shared_ptr<Table> &table, uint32_t table_count) : table_(table)  {
    active_table_ = new MemDataTable(table);
    active_table_->SetLSN(table_->GetLSN());
    for (int i = 1; i < table_count; i++) {
        free_tables_.emplace_back(new MemDataTable(table));
    }
    chunk_io_ = std::make_shared<DataChunkIO>(table->GetName());
    flush_thread_ = std::make_shared<std::thread>(FlushTable, this);

    // init indices
    for (auto col : table->GetAllColumn()) {
        if (col.GetIndexType() == LINEAR_INDEX) {
            linear_indices_.insert({col.GetName(), std::make_shared<LinearIndex>(table->GetName(), col)});
        }
    }
}

bool DataTableManager::AppendTuple(Tuple& tuple) {
    MemDataTable* table = GetActiveMemTable();
    table->AppendTuple(tuple);
    if (table->IsFull()) {
        FreezeMemTable(table);
    }
}

void DataTableManager::Query(const QueryPhrase& query, std::vector<Tuple>& res) {
    // first query active tables
    active_table_->Query(query, res);
    std::cout << "n records size in active_table: " << res.size() << std::endl;
    // then query frozen table
    // TODO concurrent lock
//    for (MemDataTable* table : frozen_tables_) {
//        table->Query(query, res);
//    }
    // query disk records, Sparse index
    QueryDiskChunk(query, res);
}

void DataTableManager::QueryDiskChunk(QueryPhrase query, std::vector<Tuple> &res) {
    std::vector<Selection> selections = query.GetSelections();
    std::vector<std::string> query_columns = query.GetQueryColumns();
    std::vector<int> column_offsets;
    for (const Selection& sel : selections) {
        int offset = table_->GetColumnIndex(sel.column_name);
        column_offsets.push_back(offset);
    }

    std::vector<int> query_column_offsets;
    for (const std::string& column : query_columns) {
        int offset = table_->GetColumnIndex(column);
        if (offset < 0) {
            continue;
        }
        query_column_offsets.push_back(offset);
    }

    int chunk_id = 0;
    int n_read_chunk = 0;
    int n_total_chunk = 0;
    while (true) {
        bool valid_chunk = true;
        bool null_chunk = false;
        n_total_chunk++;
        for (int j = 0; j < column_offsets.size(); j++) {
            int offset = column_offsets[j];
            ChunkDesc desc;
            if (!chunk_io_->ReadDataChunkDesk(chunk_id+offset, desc)) {
                null_chunk = true;
                break;
            }
            // filter by sparse index
            switch (selections[j].comp) {
            case EQUAL:
                if (desc.min_val.Compare(GREATER_THAN, selections[j].val) == CmpBool::CmpTrue ||
                desc.max_val.Compare(LESS_THAN, selections[j].val) == CmpBool::CmpTrue) {
                    valid_chunk = false;
                }
                break;
            case LESS_THAN:
                if (desc.min_val.Compare(LESS_THAN, selections[j].val) == CmpBool::CmpFalse) {
                    valid_chunk = false;
                }
                break;
            case GREATER_THAN:
                if (desc.max_val.Compare(GREATER_THAN, selections[j].val) == CmpBool::CmpFalse) {
                    valid_chunk = false;
                }
                break;
            case LESS_EQUAL:
                if (desc.min_val.Compare(LESS_EQUAL, selections[j].val) == CmpBool::CmpFalse) {
                    valid_chunk = false;
                }
                break;
            case GREATER_EQUAL:
                if (desc.max_val.Compare(GREATER_EQUAL, selections[j].val) == CmpBool::CmpFalse) {
                    valid_chunk = false;
                }
                break;
            }
            if (query.Or_Operation()) {
                if (valid_chunk) {
                    break;
                } else if (j < column_offsets.size()-1) {
                    continue;
                }
            }
            if (!valid_chunk) {
                break;
            }
        }
        if (null_chunk) {
            break;
        }

        if (!valid_chunk) {
            chunk_id += table_->GetColumnCount();
            continue;
        }
        n_read_chunk++;

        // Fetch all need chunks
        std::vector<DataChunk*> chunk_vec;
        {
            int i, j;
            i = j = 0;
            std::sort(column_offsets.begin(), column_offsets.end());
            std::sort(query_column_offsets.begin(), query_column_offsets.end());
            int next_index = 0;
            while (next_index < table_->GetColumnCount()) {
                int min_column_offset = table_->GetColumnCount();
                int min_query_offset = table_->GetColumnCount();
                if (i < column_offsets.size()) {
                    min_column_offset = column_offsets[i];
                }
                if (j < query_column_offsets.size()) {
                    min_query_offset = query_column_offsets[j];
                }
                while (next_index < std::min(i, j)) {
                    chunk_vec.push_back(nullptr);
                    next_index++;
                }
                if (next_index == min_column_offset && next_index == min_query_offset) {
                    chunk_vec.push_back(chunk_io_->ReadDataChunk(chunk_id + column_offsets.front()));
                    i++;
                    j++;
                } else if (next_index == min_column_offset) {
                    chunk_vec.push_back(chunk_io_->ReadDataChunk(chunk_id + column_offsets.front()));
                    i++;
                } else {
                    chunk_vec.push_back(chunk_io_->ReadDataChunk(chunk_id + query_column_offsets.front()));
                    j++;
                }
                next_index++;
            }
        }

        for (int i = 0; i < RECORD_COUNT_PER_CHUNK; i++) {
            bool valid = true;
            for (int j = 0; j < column_offsets.size(); j++) {
                int offset = column_offsets[j];
                if (chunk_vec[offset]->QueryIndex(i).Compare(selections[j].comp, selections[j].val) == CmpBool::CmpFalse) {
                    valid = false;
                }
            }
            if (!valid) {
                continue;
            }
            std::vector<Value> query_values;
            query_values.reserve(query_column_offsets.size());
            for (int offset : query_column_offsets) {
                query_values.push_back(chunk_vec[offset]->QueryIndex(i));
            }
            res.emplace_back(Tuple(query_values));
        }

        // release data chunk
        for (DataChunk* chunk : chunk_vec) {
            chunk_io_->UnPin(chunk->GetChunkId());
        }
        chunk_id += table_->GetColumnCount();
    }

    std::cout << n_total_chunk << " " << n_read_chunk << std::endl;
}

void DataTableManager::InsertMemTable(MemDataTable *mem_table) {
    std::lock_guard<std::mutex> lck(free_queue_mtx_);
    free_tables_.push_back(mem_table);
    free_queue_cv_.notify_one();
}

DataTableManager::~DataTableManager() {
    std::unique_lock<std::mutex> frozen_lck(frozen_queue_mtx_);
    frozen_tables_.push_back(nullptr);
    frozen_lck.unlock();
    flush_cv_.notify_one();

    flush_thread_->join();
    delete active_table_;
    while (!free_tables_.empty()) {
        delete free_tables_.front();
        free_tables_.pop_front();
    }
    timer_.StopAndPrint("Manager append records total.");
}

bool DataTableManager::DeleteTuple(Tuple &tuple) {
    return false;
}

bool DataTableManager::AppendManyTuple(std::vector<Tuple> &tuples) {
    timer_.Start();
    MemDataTable* table = GetActiveMemTable();
//    for (auto& tuple : tuples) {
//        if (table->HasFreeRecordSlot()) {
//            table->AppendTuple(tuple);
//        } else {
//            FreezeMemTable(table);
//            table = GetActiveMemTable();
//        }
//    }
    int tuple_size = tuples.size();
    int next_append = 0;
    while (next_append < tuple_size) {
        int rest_count = tuple_size-next_append;
        int free_count = table->GetFreeRecordSlot();
        int append_size = free_count > rest_count ? rest_count : free_count;
        table->AppendManyTuples(tuples, next_append, next_append+append_size-1);
        if (!table->HasFreeRecordSlot()) {
            FreezeMemTable(table);
            table = GetActiveMemTable();
        }
        next_append += append_size;
    }
    timer_.PauseAndPrint("Manager Append records.");
}

MemDataTable *DataTableManager::GetActiveMemTable() {
    if (active_table_ == nullptr) {
        std::unique_lock <std::mutex> lck(free_queue_mtx_);
        while (free_tables_.empty()) {
            free_queue_cv_.wait(lck);
        }
        active_table_ = free_tables_.front();
        free_tables_.pop_front();

        table_->IncreaseLSN();
        active_table_->SetLSN(table_->GetLSN());
    }
    return active_table_;
}

void DataTableManager::FreezeMemTable(MemDataTable *table) {
    std::unique_lock<std::mutex> frozen_lck(frozen_queue_mtx_);
    frozen_tables_.push_back(table);
    frozen_lck.unlock();
    flush_cv_.notify_one();

    if (active_table_ == table) {
        active_table_ = nullptr;
    }
}

void DataTableManager::FreezeActiveMemTable() {
    FreezeMemTable(active_table_);
    active_table_ = nullptr;
}

void DataTableManager::QueryPointCloud(PointCloudQuery& query, PointCloud<PointXYZ>& res) {
    // first query active tables
    active_table_->QueryPointCloud(query, res);
    std::cout << "n records size in active_table: " << res.Size() << std::endl;
    // then query frozen table
    // TODO concurrent lock
    for (MemDataTable* table : frozen_tables_) {
        table->QueryPointCloud(query, res);
    }
    // query disk records, Sparse index
    QueryPointCloudFromDiskChunk(query, res);
}

void DataTableManager::QueryPointCloudFromDiskChunk(PointCloudQuery &query, PointCloud<PointXYZ> &res) {
    int chunk_id = 0;
    int n_read_chunk = 0;
    int n_total_chunk = 0;
    while (true) {
        ChunkDesc x_chunk_desc;
        ChunkDesc y_chunk_desc;
        if (!chunk_io_->ReadDataChunkDesk(chunk_id+2, x_chunk_desc)) {
            break;
        }
        if (!chunk_io_->ReadDataChunkDesk(chunk_id+3, y_chunk_desc)) {
            break;
        }
        n_total_chunk++;
        auto min_x = x_chunk_desc.min_val.GetAs<float>();
        auto max_x = x_chunk_desc.max_val.GetAs<float>();
        auto min_y = y_chunk_desc.min_val.GetAs<float>();
        auto max_y = y_chunk_desc.max_val.GetAs<float>();

        if (query.MaySatisfyIn(min_x, max_x, min_y, max_y)) {
            n_read_chunk++;
            // Fetch all need chunks
            auto x_chunk = chunk_io_->ReadDataChunk(chunk_id+2);
            auto y_chunk = chunk_io_->ReadDataChunk(chunk_id+3);
            auto z_chunk = chunk_io_->ReadDataChunk(chunk_id+4);
            for (uint i = 0; i < RECORD_COUNT_PER_CHUNK; i++) {
                auto x = x_chunk->QueryIndex(i).GetAs<float>();
                auto y = y_chunk->QueryIndex(i).GetAs<float>();
                if (!query.Satisfy(x, y)) {
                    continue;
                }
                auto z = z_chunk->QueryIndex(i).GetAs<float>();
                res.Append({i, x, y, z});
            }

            // release data chunk
            chunk_io_->UnPin(x_chunk->GetChunkId());
            chunk_io_->UnPin(y_chunk->GetChunkId());
            chunk_io_->UnPin(z_chunk->GetChunkId());
        }

        chunk_id += table_->GetColumnCount();
//        LOG_DEBUG("n_total_chunk %d", n_total_chunk);
    }

    std::cout << "total_chunk: " << n_total_chunk << "; read_chunk: " << n_read_chunk << std::endl;
}

bool DataTableManager::AppendPoint(PointXYZI &point) {
    MemDataTable* table = GetActiveMemTable();;
    table->AppendPointData(reinterpret_cast<const char *>(&point), sizeof(point));
    if (!table->HasFreeRecordSlot()) {
        FreezeMemTable(table);
    }
}

}